package org.delay;

import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Classname DelayQueueMessageConsumer
 * @Description
 * @Date 2022/9/19 11:08
 * @Created by wangchangjiu
 */
@Slf4j
public abstract class DelayQueueMessageConsumer {

    private DelayQueueMessage delayQueueMessage;

    private DelayQueueTopicRegistrar delayQueueTopicRegistrar;

    private RedissonDelayProperties properties;

    public DelayQueueMessageConsumer(DelayQueueMessage delayQueueMessage,
                                          DelayQueueTopicRegistrar delayQueueTopicRegistrar,
                                          RedissonDelayProperties properties){
        this.delayQueueMessage = delayQueueMessage;
        this.delayQueueTopicRegistrar = delayQueueTopicRegistrar;
        this.properties = properties;
    }



    @PostConstruct
    public void consumer() throws Exception {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(() -> {
            while (true) {
                Set<String> topics = delayQueueTopicRegistrar.pullAll();
                if(!CollectionUtils.isEmpty(topics)){
                    topics.stream().forEach(topic -> {
                        Object value = delayQueueMessage.getDelayQueue(topic);
                        if (value != null) {
                            // 得到消息 再发送到 kafka 消息去消费
                            processMessages(topic, value);
                        }
                    });
                }
                Thread.sleep(properties.getConsumerSleep());
            }
        });
        log.info("(Redis延迟队列启动成功)");
    }


    public abstract void processMessages(String topic, Object value);
}
